package com.ds.infrastructure.hyperspace.container.endpoint.destination;

import com.ds.infrastructure.hyperspace.common.Endpoint;
import com.ds.infrastructure.hyperspace.common.EndpointNode;
import com.ds.infrastructure.hyperspace.common.config.TargetNodeConfig;
import com.ds.infrastructure.hyperspace.container.endpoint.connector.HyperspaceConnector;
import com.ds.infrastructure.hyperspace.container.endpoint.connector.health.HyperspaceNodeHealthChecker;
import com.ds.infrastructure.hyperspace.container.endpoint.destination.resolver.ResolverProvider;
import com.ds.infrastructure.hyperspace.container.endpoint.destination.resolver.dns.DnsResolverProvider;
import com.ds.infrastructure.hyperspace.container.endpoint.destination.resolver.k8s.K8sResolverProvider;
import com.ds.infrastructure.hyperspace.container.endpoint.destination.resolver.raw.RawResolverProvider;
import com.ds.infrastructure.hyperspace.container.flow.context.HyperspaceDuplexFlowContext;
import com.ds.infrastructure.hyperspace.container.flow.operation.TargetFlowOperation;
import com.ds.infrastructure.hyperspace.container.internal.NettyContainer;
import com.ds.infrastructure.hyperspace.container.internal.util.AddressesUtil;
import io.netty.channel.EventLoop;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;

import java.net.SocketAddress;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import static com.ds.infrastructure.hyperspace.common.config.TargetNodeConfig.HTTPS_DEFAULT_PORT;
import static com.ds.infrastructure.hyperspace.common.config.TargetNodeConfig.HTTP_DEFAULT_PORT;
import static com.ds.infrastructure.hyperspace.container.internal.util.StringConstant.COLON;

/**
 * @author  Weishaoying
 */
@Slf4j
public class TargetEndpointNode implements EndpointNode<TargetNodeConfig> {

    private final NettyContainer nettyContainer;

    protected final TargetFlowOperation<HyperspaceDuplexFlowContext> targetFlowOperation;

    private final Endpoint endpoint;

    private volatile TargetNodeConfig targetNodeConfig;

    @Getter
    private volatile String key;

    private AtomicInteger failCount = new AtomicInteger();

    private AtomicInteger passCount = new AtomicInteger();

    private AtomicBoolean available = new AtomicBoolean(false);

    private HyperspaceNodeHealthChecker healthChecker;

    @Getter
    private volatile ResolverProvider resolverProvider;

    @Getter
    private volatile boolean isShutdown;

    public TargetEndpointNode(NettyContainer nettyContainer, TargetNodeConfig targetNodeConfig, Endpoint endpoint) {
        this.nettyContainer = nettyContainer;
        this.targetFlowOperation = nettyContainer.getFlowPipelineOrchestrator().getTargetFlowOperation();
        this.targetNodeConfig = targetNodeConfig;
        this.endpoint = endpoint;
        this.key = TargetEndpointNode.getNodeKey(endpoint.getNamespace(), targetNodeConfig);
        this.healthChecker = new HyperspaceNodeHealthChecker(nettyContainer, this);
        this.resolverProvider = newResolverProvider(nettyContainer, targetNodeConfig, endpoint.getNamespace());
    }

    private ResolverProvider newResolverProvider(NettyContainer nettyContainer, TargetNodeConfig targetNodeConfig, String namespace) {
        switch (targetNodeConfig.getResolverType()) {
            case RAW:
                return new RawResolverProvider(nettyContainer, targetNodeConfig, namespace);
            case DNS:
                return new DnsResolverProvider(nettyContainer, targetNodeConfig, namespace);
            case K8S:
            case K8SS:
                return new K8sResolverProvider(nettyContainer, targetNodeConfig, namespace);
            default:
                log.warn("Target node({}) use default raw resolver because unsupported resolver type:{}. ", getKey(), targetNodeConfig.getResolverType());
                return new RawResolverProvider(nettyContainer, targetNodeConfig, namespace);
        }
    }

    @Override
    public void incrementFailCount(SocketAddress remoteAddress) {
        if (remoteAddress == null || !resolverProvider.isAvailable(remoteAddress)) {
            return;
        }
        addTargetNodeStatusChangeMetric(remoteAddress, false);
        if (failCount.incrementAndGet() < getMaxFailCount()) {
            return;
        }
        failCount.set(0);
        setAvailable(false);
    }

    private int getMaxFailCount() {
        return targetNodeConfig.getProtocolConfig().getHealthCheckConfig().getFails() * resolverProvider.getAllNode().size();
    }

    @Override
    public void incrementPassCount(SocketAddress remoteAddress) {
        if (!resolverProvider.isAvailable(remoteAddress)) {
            return;
        }
        addTargetNodeStatusChangeMetric(remoteAddress, true);
        if (passCount.incrementAndGet() < getMinPassCount()) {
            return;
        }
        passCount.set(0);
        if (!available.get()) {
            setAvailable(true);
            Map<EventLoop, HyperspaceConnector> hyperspaceConnectorMap =
                    nettyContainer.getHyperspaceConnectorMap();
            for (HyperspaceConnector hyperspaceConnector : hyperspaceConnectorMap.values()) {
                hyperspaceConnector.eventLoop().submit(() -> {
                    try {
                        hyperspaceConnector.recoverNodeStreamPool(this);
                    } catch (Exception e) {
                        log.error("Recover node stream pool failed!", e);
                    }
                });
            }
        }
    }

    private int getMinPassCount() {
        return targetNodeConfig.getProtocolConfig().getHealthCheckConfig().getPasses();
    }

    @Override
    public boolean isAvailable() {
        return available.get() && !isEmptyNodes();
    }

    @Override
    public TargetNodeConfig getNodeConfig() {
        return targetNodeConfig;
    }

    @Override
    public String getHostHeader() {
        int nodePort = targetNodeConfig.getPort();
        if (nodePort == HTTP_DEFAULT_PORT || nodePort == HTTPS_DEFAULT_PORT) {
            return targetNodeConfig.getHost();
        } else {
            return targetNodeConfig.getHost() + COLON + nodePort;
        }
    }

    @Override
    public String getHostInfo() {
        return new StringBuilder().append(targetNodeConfig.getHost())
                .append(COLON)
                .append(targetNodeConfig.getPort()).toString();
    }

    @Override
    public void updateNodeConfig(TargetNodeConfig targetNodeConfig) {
        this.targetNodeConfig = targetNodeConfig;
        this.key = TargetEndpointNode.getNodeKey(endpoint.getNamespace(), targetNodeConfig);
        if (healthChecker != null) {
            healthChecker.reschedule();
        }
    }

    @Override
    public Endpoint getEndpoint() {
        return endpoint;
    }

    @Override
    public void shutdown() {
        this.isShutdown = true;
        if (healthChecker != null) {
            healthChecker.cancel();
        }
        if (resolverProvider != null) {
            resolverProvider.shutdown();
        }
    }

    @Override
    public boolean isEmptyNodes() {
        return CollectionUtils.isEmpty(resolverProvider.getAllNode());
    }

    public static String getNodeKey(String endpointNamespace, TargetNodeConfig targetNodeConfig) {
        return new StringBuilder().append(endpointNamespace).append(COLON)
                .append(targetNodeConfig.getHost()).append(COLON)
                .append(targetNodeConfig.getPort()).append(COLON)
                .append(targetNodeConfig.getZone()).append(COLON)
                .append(targetNodeConfig.getResolverType()).toString();
    }

    private void setAvailable(boolean available) {
        this.available.compareAndSet(!available, available);
    }

    private void addTargetNodeStatusChangeMetric(SocketAddress remoteAddress, boolean available) {
        if (remoteAddress == null) {
            return;
        }
        targetFlowOperation.handleTargetNodeStatusChange(
                endpoint.getNamespace(), AddressesUtil.getIpAndPort(remoteAddress), available ? 1 : 0);
    }

}
